from abc import ABC, abstractmethod
from typing import List
from io import BytesIO
import pandas as pd
import time
import base64
import pydantic
import sqlalchemy as sa
from fastapi_pagination.bases import AbstractPage
from fastapi_pagination.ext.databases import paginate
from sqlalchemy.dialects import mysql
from like.dependencies.database import db
from like.utils.urls import UrlUtil
from like.utils.array import ArrayUtil
from .models import test
from .schemas import (
    TestListIn, TestExportIn, TestAddIn, TestEditIn, TestOut)


class ITestService(ABC):
    """测试抽象类"""

    @abstractmethod
    async def listBase(self, list_in: TestListIn) -> sa.sql.Select:
        pass

    @abstractmethod
    async def list(self, list_in: TestListIn) -> AbstractPage[TestOut]:
        pass

    @abstractmethod    
    async def export_excel(self, export_in: TestExportIn) -> str:
        pass    

    @abstractmethod
    async def detail(self, id_: int) -> TestOut:
        pass

    @abstractmethod
    async def add(self, add_in: TestAddIn):
        pass

    @abstractmethod
    async def edit(self, edit_in: TestEditIn):
        pass

    @abstractmethod
    async def delete(self, id_: int):
        pass


class TestService(ITestService):
    """测试基类"""

    async def listBase(self, list_in: TestListIn) -> sa.sql.Select:
        """测试基类"""
        columns = [
            test,
            
            ]
        where = []
        if list_in.sex is not None and str(list_in.sex):
            where.append(test.c.sex == list_in.sex)
        if list_in.name is not None and str(list_in.name):
            where.append(test.c.name.like(f'%{ list_in.name }%'))
        if list_in.sort is not None and str(list_in.sort):
            where.append(test.c.sort == list_in.sort)
        if list_in.is_show is not None and str(list_in.is_show):
            where.append(test.c.is_show == list_in.is_show)
        where.append(test.c.is_delete == 0)
        query = sa.select(columns).where(*where) \
            .select_from(
            test
        ).order_by(test.c.sort.asc(), test.c.id.asc())
        return query

    async def list(self, list_in: TestListIn) -> AbstractPage[TestOut]:
        """测试列表"""
        query = await self.listBase(list_in)
        res = await paginate(db, query)
        return res    

    async def export_excel(self, export_in: TestExportIn) -> str:    
        """测试导出Excel"""
        result = await self.listBase(export_in)
        # 计算要导出的记录范围
        offset = (export_in.page_start - 1) * export_in.page_size
        limit = (export_in.page_end - export_in.page_start + 1) * export_in.page_size

        # 修改查询以仅选择所需的记录
        result = result.offset(offset).limit(limit)
        print(result.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True}))  # 打印完整的 SQL 查询语句
        query_result = await db.fetch_all(result)

        # 将查询结果转换为 Pandas DataFrame
        df = pd.DataFrame(query_result, columns=result.selected_columns.keys())

        # 将 DataFrame 转换为 Excel 文件的字节流
        excel_data = BytesIO()
        df.to_excel(excel_data, index=False)
        excel_data.seek(0)  # 将指针移回文件开头

        excel_bytes = excel_data.read()  # 读取字节流内容到字节字符串中
        return base64.b64encode(excel_bytes).decode('utf-8')  # 将字节字符串转换为Base64编码的字符串表示ArrayBuffer


    async def detail(self, id_: int) -> TestOut:
        """测试详情"""
        model = await db.fetch_one(
            test.select().where(
                test.c.id == id_, test.c.is_delete == 0)
            .limit(1))
        assert model, '数据不存在!'
        res = TestOut.from_orm(model)
        return res

    async def add(self, add_in: TestAddIn):
        """测试新增"""
        await db.execute(test.insert().values({
            'sex': add_in.sex,
            'name': add_in.name,
            'sort': add_in.sort,
            'is_show': add_in.is_show,
        }))

    async def edit(self, edit_in: TestEditIn):
        """测试编辑"""
        assert await db.fetch_one(
            test.select().where(
                test.c.id == edit_in.id, test.c.is_delete == 0)
            .limit(1)), '数据不存在!'
        await db.execute(
            test.update().where(test.c.id == edit_in.id).values({
                'sex': edit_in.sex,
                'name': edit_in.name,
                'sort': edit_in.sort,
                'is_show': edit_in.is_show,
            }))

    async def delete(self, id_: int):
        """测试删除"""
        assert await db.fetch_one(
            test.select().where(
                test.c.id == id_, test.c.is_delete == 0)
            .limit(1)), '数据不存在!'
        await db.execute(
            test.update().where(test.c.id == id_).values({
                'is_delete': 1,
                'delete_time': int(time.time()),
            }))
        
    @classmethod
    async def instance(cls):
        """实例化"""
        return cls()
